package net.xuele.learn.flink;

import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * 基于流窗口的单词统计应用程序，计算 5 秒窗口内来自 Web 套接字的单词数
 *
 * 自定义DataSource：
 *         1、实现 SourceFunction 接口编写自定义的非并行 source
 *         2、实现 ParallelSourceFunction 接口或者继承 RichParallelSourceFunction 类编写自定义的并行 sources
 */
public class WindowWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 接收来自网络的数据，
        DataStream<Tuple2<String, Integer>> ds = env.socketTextStream("localhost", 9999)
                // map则是一对一，次数是具体的收集逻辑
                .flatMap(new Splitter())
                // 按照单词分组
                .keyBy(e -> e.f0)
                // 窗口
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                // sum the second field of the tuple，也可以传递具体的field值
                .sum(1);
        // sink
        ds.print();


        // 在终端使用 netcat 启动输入流，实现源源不断的数据流
        env.execute();
    }

    /**
     * 根据特殊的标识分割网络上接收到字符串
     */
    private static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        private static final long serialVersionUID = 1434517788592276689L;

        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            for (String word : s.split(" ")) {
                collector.collect(new Tuple2<>(word, 1));
            }
        }
    }
}
